package rocketmq.base;

import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.RedisTemplate;
import rocketmq.constant.RedisConstant;
import rocketmq.domain.RocketMqMessage;

import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;


/**
 * rocketmq 通过判断是否抛出异常来决定是否ack 故想要重试的话直接抛异常就好了
 *
 * @author mercury
 */
public abstract class BaseListener implements RocketMQListener<RocketMqMessage> {


    protected static Logger logger = LoggerFactory.getLogger(BaseListener.class);

    public static final String REDIS_LOCK = "lock";

    public static final Integer REDIS_TIME_OUT = 25;


    @Resource
    private RedisTemplate<String, String> redisTemplate;

    @Resource
    private RocketMQProperties rocketMqProperties;

    /**
     * 业务执行方法
     *
     * @param content 消息体
     */
    protected abstract void execute(RocketMqMessage content);

    /**
     * 失败执行
     *
     * @param content 消息体
     */
    protected abstract void failExecute(RocketMqMessage content);


    protected void receiveMessage(RocketMqMessage message) {
        if (canConsume(message)) {
            //已经消费，直接返回
            logger.info("[{}] 已经消费过", message.getId());
        }
        //执行业务
        execute(message);
        logger.info("[{}] 消费成功", message.getId());
    }


    /**
     * 是否能消费，防止重复消费
     *
     * @param content 消息体
     * @return 结果
     */
    private Boolean canConsume(RocketMqMessage content) {
        if (redisTemplate.opsForValue().get(RedisConstant.queueKey(content.getId())) == null) {
            return false;
        } else {
            //存储消费标志
            redisTemplate.opsForValue().set(RedisConstant.queueKey(content.getId()), REDIS_LOCK, REDIS_TIME_OUT, TimeUnit.SECONDS);
            return true;
        }
    }

}
